package com.gamejelly.broker.server;

import com.alipay.remoting.ConnectionEventProcessor;
import com.alipay.remoting.ConnectionEventType;
import com.alipay.remoting.rpc.RpcServer;
import com.alipay.remoting.rpc.protocol.UserProcessor;
import com.gamejelly.broker.message.RegisterMsg;
import com.gamejelly.broker.message.RpcMessage;
import io.netty.util.internal.StringUtil;
import org.springframework.core.Ordered;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

public class BrokerServerImpl implements BrokerServer, Ordered {

    /**
     * bolt server
     */
    RpcServer server;

    /**
     * cli-merge-id address
     */
    final Map<String, String> cliMap = new ConcurrentHashMap<>();


    @Override
    public String register(String remoteAddress, RegisterMsg registerMsg) {

        String rpcId = registerMsg.getRpcId();

        cliMap.put(rpcId, remoteAddress);

        return rpcId;
    }


    @Override
    public void send(RpcMessage rpcMessage) throws Exception {
        String destination = cliMap.get(rpcMessage.findTarget());
        if (!StringUtil.isNullOrEmpty(destination)) {
            server.oneway(destination, rpcMessage);
        }
    }


    @Override
    public void unRegister(String rpcId) {

        cliMap.remove(rpcId);

    }

    private BrokerServerImpl aware(Collection<?> collection) {
        for (Object o : collection) {
            if (o instanceof BrokerServerAware)
                ((BrokerServerAware) o).setServer(this);
        }
        return this;
    }

    public static ServerBuilder Builder() {
        return new ServerBuilder();
    }

    private BrokerServerImpl() {
    }

    @Override
    public int getOrder() {
        return Ordered.HIGHEST_PRECEDENCE;
    }

    public static class ServerBuilder {

        int port;

        List<UserProcessor<?>> processorList = new ArrayList<>();
        Map<ConnectionEventType, ConnectionEventProcessor> eventProcessorMap = new HashMap<>();


        public ServerBuilder port(int port) {
            this.port = port;
            return this;
        }

        public ServerBuilder addMessageProcessor(UserProcessor<?> processor) {
            processorList.add(processor);
            return this;
        }

        public ServerBuilder addConnEventProcessor(ConnectionEventType eventType, ConnectionEventProcessor processor) {
            eventProcessorMap.put(eventType, processor);
            return this;
        }

        public BrokerServerImpl Build() {
            BrokerServerImpl bserver = new BrokerServerImpl();
            RpcServer rpcServer = new RpcServer(port, true);
            bserver.server = rpcServer;
            bserver.aware(processorList).aware(eventProcessorMap.values());
            processorList.forEach(rpcServer::registerUserProcessor);
            eventProcessorMap.forEach(rpcServer::addConnectionEventProcessor);
            rpcServer.startup();
            return bserver;
        }

    }

}
